package com.eshop.storm;

import com.eshop.storm.bolt.LogParseBolt;
import com.eshop.storm.bolt.ProductCountBolt;
import com.eshop.storm.cons.ProjectCons;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.BrokerHosts;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.ZkHosts;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class EshopStormApplication {

    public static void main(String[] args) {
        SpringApplication.run(EshopStormApplication.class, args);
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        // 设置kafka的zookeeper集群
        BrokerHosts hosts = new ZkHosts(ProjectCons.REMOTE_ZK);
        // 初始化配置信息
        SpoutConfig spoutConfig = new SpoutConfig(hosts, ProjectCons.TOPIC, ProjectCons.ZK_ROOT, "AccessLogKafkaSpout");
        // 在topology中设置spout
        topologyBuilder.setSpout("AccessLogKafkaSpout", new KafkaSpout(spoutConfig), 1);
        topologyBuilder.setBolt("LogParseBolt", new LogParseBolt(), 1).setNumTasks(1).shuffleGrouping("AccessLogKafkaSpout");
        topologyBuilder.setBolt("ProductCountBolt", new ProductCountBolt(), 1).setNumTasks(1).fieldsGrouping("LogParseBolt", new Fields("productId"));
        Config config = new Config();
        if (args != null && args.length > 0) {
            // 集群模式
            config.setNumWorkers(3);
            try {
                StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology());
            } catch (Exception e) {
                e.printStackTrace();
            }
        } else {
            // 本地模式
            config.setDebug(true);
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("HotProductTopology", config, topologyBuilder.createTopology());
            // Utils.sleep(100000);
            // localCluster.shutdown();
        }
    }
}
